package com.zb;

import com.ctrip.framework.apollo.Config;
import com.ctrip.framework.apollo.ConfigService;
import com.ctrip.framework.apollo.model.ConfigChangeEvent;
import io.micrometer.core.instrument.MeterRegistry;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.*;

@Slf4j
public class ThreadPoolFactory {

    private final Map<String, ExecutorService> executorServiceMap = new ConcurrentHashMap<>();
    private final ThreadPoolProperties threadPoolProperties;
    private final MeterRegistry meterRegistry;
    private final Config apolloConfig;


    public ThreadPoolFactory(ThreadPoolProperties threadPoolProperties, MeterRegistry meterRegistry) {
        this.threadPoolProperties = threadPoolProperties;
        this.meterRegistry = meterRegistry;
        this.apolloConfig = ConfigService.getConfig(threadPoolProperties.getApolloNamespace());
    }

    public ExecutorService build(String threadPoolName) {
        return executorServiceMap.computeIfAbsent(threadPoolName, this::createThreadPool);
    }


    private ExecutorService createThreadPool(String threadPoolName) {
        // 根据CPU计算默认的线程池大小配置
        ThreadPoolConfig threadPoolConfig = new ThreadPoolConfig();
        threadPoolConfig.setThreadPoolName(threadPoolName);
        threadPoolConfig.setCoreSize(CpuCoreUtils.getDefaultCoreSize(threadPoolName));
        threadPoolConfig.setMaxSize(CpuCoreUtils.getDefaultMaxSize(threadPoolName));
        threadPoolConfig.setQueueCapacity(CpuCoreUtils.getDefaultQueueCapacity(threadPoolProperties.getDefaultQueueCapacity()));
        threadPoolConfig.setRejectedPolicy(new ThreadPoolExecutor.CallerRunsPolicy());

        // 加载Apollo配置 重写线程池数据
        this.loadApolloConfig(threadPoolConfig);

        // 加载线程池配置
        ThreadPoolExecutor executor = loadThreadPoolExecutor(threadPoolConfig);

        // 注册监控
        if (meterRegistry != null) {
            meterRegistry.gauge("thread.pool.active", executor, ThreadPoolExecutor::getActiveCount);
            meterRegistry.gauge("thread.pool.queue.size", executor.getQueue(), java.util.Queue::size);
        }
        registerApolloListener(threadPoolName);

        log.info("线程池[{}]初始化完成: core={}, max={}, queue={}",
                threadPoolName, threadPoolConfig.getCoreSize(), threadPoolConfig.getMaxSize(), threadPoolConfig.getQueueCapacity());
        return executor;
    }

    private void loadApolloConfig(ThreadPoolConfig config) {
        String prefix = config.getThreadPoolName() + ".";
        config.setCoreSize(apolloConfig.getIntProperty(prefix + "coreSize", config.getCoreSize()));
        config.setMaxSize(apolloConfig.getIntProperty(prefix + "maxSize", config.getMaxSize()));
        config.setQueueCapacity(apolloConfig.getIntProperty(prefix + "queueCapacity", config.getQueueCapacity()));
        config.setKeepAliveSeconds(apolloConfig.getIntProperty(prefix + "keepAliveSeconds", 60));
        config.setThreadNamePrefix(apolloConfig.getProperty(prefix + "threadNamePrefix", config.getThreadNamePrefix()));

        String policy = apolloConfig.getProperty(prefix + "rejectedPolicy", "AbortPolicy");
        config.setRejectedPolicy(RejectedPolicyRegistry.getPolicy(policy));
    }

    private ThreadPoolExecutor loadThreadPoolExecutor(ThreadPoolConfig threadPoolConfig) {

        return new ThreadPoolExecutor(
                threadPoolConfig.getCoreSize(),
                threadPoolConfig.getMaxSize(),
                threadPoolConfig.getKeepAliveSeconds(),
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(threadPoolConfig.getQueueCapacity()),
                r -> new Thread(r, threadPoolConfig.getThreadNamePrefix()),
                threadPoolConfig.getRejectedPolicy());

    }

    // 注册Apollo监听器 监听配置变化 重新创建线程池
    private void registerApolloListener(String threadPoolName) {
        apolloConfig.addChangeListener((ConfigChangeEvent event) -> {
            String prefix = threadPoolName + ".";
            if (event.changedKeys().stream().anyMatch(key -> key.startsWith(prefix))) {
                log.info("线程池[{}]配置变更，重建中...", threadPoolName);
                ExecutorService newExecutor = createThreadPool(threadPoolName);
                replaceThreadPool(threadPoolName, newExecutor);
            }
        });
    }

    private void replaceThreadPool(String threadPoolName, ExecutorService newExecutor) {
        ExecutorService oldExecutor = executorServiceMap.get(threadPoolName);
        if (oldExecutor == null) {
            executorServiceMap.put(threadPoolName, newExecutor);
            return;
        }

        executorServiceMap.replace(threadPoolName, newExecutor);
        new Thread(() -> {
            try {
                oldExecutor.shutdown();
                if (!oldExecutor.awaitTermination(5, TimeUnit.MINUTES)) {
                    oldExecutor.shutdownNow();
                    log.warn("线程池[{}]强制关闭超时任务", threadPoolName);
                }
            } catch (InterruptedException e) {
                oldExecutor.shutdownNow();
            }
        }).start();
    }
}
